package com.bjtu.boatmq.client.producer;

import java.util.List;
import java.util.Timer;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;

import com.bjtu.boatmq.common.log.Log;
import com.bjtu.boatmq.common.protocol.Message;

public class DefaultMQProducer implements MQProducer{
	private ServiceState serviceState = ServiceState.CREATE_JUST;
	private int defaultTopicQueueNums = 3;
	private int defaultTopicPartionBackupNums=3;
	private Log log;
	Router r;
	SendMsg sendutils;
    private BlockingQueue<Runnable> senderThreadPoolQueue;
    private ExecutorService senderExecutor;
    private Timer timer=new Timer();
    public DefaultMQProducer() {
    	log=new Log();
    	this.sendutils=new SendMsg();
    	this.senderThreadPoolQueue = new LinkedBlockingQueue<Runnable>(50000);
        this.senderExecutor = new ThreadPoolExecutor(
            Runtime.getRuntime().availableProcessors(),//指定了线程池中的线程数量，它的数量决定了添加的任务是开辟新的线程去执行，还是放到workQueue任务队列中去
            Runtime.getRuntime().availableProcessors(),//指定了线程池中的最大线程数量，这个参数会根据你使用的workQueue任务队列的类型，决定线程池会开辟的最大线程数量
            1000 * 60,//当线程池中空闲线程数量超过corePoolSize时，多余的线程会在多长时间内被销毁
            TimeUnit.MILLISECONDS,//keepAliveTime的单位
            this.senderThreadPoolQueue,//任务队列，被添加到线程池中，但尚未被执行的任务
            new ThreadFactory() {//线程工厂，用于创建线程
                private AtomicInteger threadIndex = new AtomicInteger(0);//并发源自类
                @Override
                public Thread newThread(Runnable r) {
                    return new Thread(r, "SenderExecutor_" + this.threadIndex.incrementAndGet());
                }
            });
    }
    
	public void start() {
		// TODO Auto-generated method stub
		switch (this.serviceState) {
	        case CREATE_JUST:
	            this.serviceState = ServiceState.START_FAILED;
	          //建立router center长连接，定期获取路由信息
	            this.senderExecutor.execute(new Runnable() {
					@Override
					public void run() {
						// TODO Auto-generated method stub
						timer.schedule(new MyTask(r),0,1000);
					}	            	
	            });
	            this.log.writeInfo("Producer start successfully");
	            this.serviceState=ServiceState.RUNNING;
	            break;
			default:
				break;
		}
	}

	@Override
	public void shutdown() {
		// TODO Auto-generated method stub
		try {
			this.timer.cancel();
			this.senderExecutor.shutdown();
			//this.log.writeInfo("Producer stop successfully");
		}catch(Exception e) {
			this.log.writeError("Producer stop error");
			this.log.writeError(e.toString());
		}
		
		
	}
	@Override
	public void send(List<Message> msg) {
		// TODO Auto-generated method stub
        try {
        	senderExecutor.submit(new Runnable() {
                @Override
                public void run() {
                        try {
                            sendDefaultImpl(msg);
                        } catch (Exception e) {
                            
                        }
                    }
                });
        } catch (Exception e) {
        	e.printStackTrace();
            this.log.writeError(e.toString());
        }
	}
	private void sendDefaultImpl(List<Message> msg) throws Exception {
		this.sendutils.send(msg, r);
	}
	@Override
	public void createTopic(String newTopic){
		
		// TODO Auto-generated method stub
		try {
			this.createTopic(newTopic, defaultTopicQueueNums);
		} catch (Exception e) {
			// TODO Auto-generated catch block
			this.log.writeError(e.toString());
			e.printStackTrace();
		}
	}
	@Override
	public void createTopic(String newTopic, int partitionNum){
		// TODO Auto-generated method stub
		try {
			this.r.createTopic(newTopic, partitionNum,defaultTopicPartionBackupNums);
		} catch (Exception e) {
			// TODO Auto-generated catch block
			this.log.writeError(e.toString());
			e.printStackTrace();
		}
	}
	public void setZKServer(String ip) {
		this.r=new Router(ip);
	}

	@Override
	public void createTopic(String newTopic, int partitionNum, int backupNum){
		// TODO Auto-generated method stub
		try {
			this.r.createTopic(newTopic, partitionNum, backupNum);
		} catch (Exception e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}
	}
	
}
